package com.vf.cloud.paas.app.master.common.server.agentManager.hander;

import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

import com.alibaba.fastjson.JSONObject;
import com.jfinal.kit.JsonKit;
import com.jfinal.kit.StrKit;
import com.jfinal.plugin.activerecord.Db;
import com.vf.cloud.paas.app.agent.common.util.KeyUtil;
import com.vf.cloud.paas.app.master.common.domain.Project;
import com.vf.cloud.paas.app.master.common.domain.RenderAgent;
import com.vf.cloud.paas.app.master.common.domain.RenderAgentGpu;
import com.vf.cloud.paas.app.master.common.domain.RenderAgentRelay;
import com.vf.cloud.paas.app.master.common.domain.RenderAgentSignalling;
import com.vf.cloud.paas.app.master.common.domain.RenderLog;
import com.vf.cloud.paas.common.cirrus.Cirrus;
import com.vf.cloud.paas.common.service.IRedis;
import com.vf.cloud.paas.common.util.DateUtil;
import com.vf.cloud.paas.common.util.IPUtil;
import com.vf.cloud.paas.common.util.PrefixUtil;
import com.vf.cloud.paas.common.vo.GpuInfo;
import com.vf.cloud.paas.common.vo.OsInfo;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.HttpHeaderNames;
import io.netty.handler.codec.http.HttpMessage;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.HttpUtil;
import io.netty.handler.codec.http.HttpVersion;
import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
import io.netty.handler.codec.http.websocketx.PingWebSocketFrame;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketServerHandshaker;
import io.netty.handler.codec.http.websocketx.WebSocketServerHandshakerFactory;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.CharsetUtil;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class AgentManagerHandler extends SimpleChannelInboundHandler<Object> {

	private IRedis redisImpl;

	public AgentManagerHandler(IRedis redisImpl) {
		super();
		this.redisImpl = redisImpl;

	}

	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception {
		log.info(String.format("RenderNoder connected:" + ctx.channel().remoteAddress()));
	}

	@Override
	public void channelInactive(ChannelHandlerContext ctx) throws Exception {
		log.error(String.format("RenderNoder disconnected:" + ctx.channel().remoteAddress()));
		onRenderDisconnected(ctx);
	}

	@Override
	public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
		ctx.flush();
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
		// log.error("RenderNoder connection error:", cause);
		ctx.close();
	}

	@Override
	protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
		if (msg instanceof FullHttpRequest) {
			handleHttpRequest(ctx, (FullHttpRequest) msg);
		} else {
			handlerWebSocketFrame(ctx, msg);
		}
	}

	protected WebSocketServerHandshaker handshaker;

	/**
	 * 处理 FullHttpRequest
	 * 
	 * @param ctx
	 * @param request
	 */
	protected void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest req) {
		if (!req.decoderResult().isSuccess() || (!"websocket".equals(req.headers().get("Upgrade")))) {
			sendHttpResponse(ctx, req,
					new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));
			return;
		}

		String uri = req.uri();
		WebSocketServerHandshakerFactory wsFactory = new WebSocketServerHandshakerFactory(
				"ws://" + req.headers().get(HttpHeaderNames.HOST) + uri, null, false);
		handshaker = wsFactory.newHandshaker(req);
		if (handshaker == null) {
			WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
		} else {
			handshaker.handshake(ctx.channel(), req);
		}

		uri = uri.replaceFirst("\\/", "");
		String[] params = uri.split("\\/", -1);
		if (params == null || params.length <= 0) {
			ctx.close();
			return;
		}

		String token = params[0];
		if (StrKit.isBlank(token)) {
			sendUnauthorized(ctx);
			ctx.close();
			return;
		}

//		String appid = params[1];
//		if (StrKit.isBlank(appid)) {
//			sendUnauthorized(ctx);
//			ctx.close();
//			return;
//		}

		RenderAgent renderAgent = RenderAgent.dao.findById(token);
		if (renderAgent == null) {
			sendUnauthorized(ctx);
			ctx.close();
			log.error(String.format("[%s]>>> Unauthorized", token));
			return;
		}

		if (!StrKit.equals(renderAgent.getStatus(), "0")) {
			sendNotAvailable(ctx);
			ctx.close();
			return;
		}

		sendConnected(token,ctx);
		Cirrus.AgentManager.put(token, ctx);
		redisImpl.add(ctx.channel().id().asLongText(), token);
	}

	public static void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest req, DefaultFullHttpResponse res) {
		if (res.status().code() != 200) {
			ByteBuf buf = Unpooled.copiedBuffer(res.status().code() + "", CharsetUtil.UTF_8);
			res.content().writeBytes(buf);
			buf.release();
		}
		// 如果是非Keep-Alive，关闭连接
		if (!HttpUtil.isKeepAlive((HttpMessage) req.protocolVersion()) || res.status().code() != 200) {
			ctx.channel().writeAndFlush(res).addListener(ChannelFutureListener.CLOSE);
		}
	}

	private void handlerWebSocketFrame(ChannelHandlerContext ctx, Object frame) {
		// 判断是否关闭链路的指令
		if (frame instanceof CloseWebSocketFrame) {
			CloseWebSocketFrame closeWebSocketFrame = (CloseWebSocketFrame) frame;
			handshaker.close(ctx.channel(), (CloseWebSocketFrame) closeWebSocketFrame.retain());
			return;
		}
		// 判断是否ping消息
		if (frame instanceof PingWebSocketFrame) {
			return;
		}

		if (frame instanceof TextWebSocketFrame) {

			TextWebSocketFrame textWebSocketFrame = (TextWebSocketFrame) frame;
			JSONObject rawMsg = null;
			try {
				rawMsg = JSONObject.parseObject(textWebSocketFrame.text());
			} catch (Exception e) {
				log.error(String.format("[%s]>> cannot parse RenderNoder message:%s", 1008, e));
				ctx.close();
			}
			if (rawMsg == null) {
				return;
			}

			String type = rawMsg.getString("type");
			if (StrKit.equals(type, "ping")) {
				sendPing(ctx, rawMsg.getLong("time"));
				return;
			}

			if (StrKit.equals(type, "GPU_Info")) {
				onGPUInfo(ctx, rawMsg);
			} else if (StrKit.equals(type, "GPU_Init")) {
				onGPUInit(ctx, rawMsg);
			} else if (StrKit.equals(type, "onStartClient")) {
				if (Cirrus.orders.containsKey(rawMsg.getString("uuid"))) {
					Cirrus.orders.put(rawMsg.getString("uuid"), rawMsg.getString("r"));
				}

			} else if (StrKit.equals(type, "Os_Info")) {
				
				StringBuffer sb=new StringBuffer();
				
				OsInfo osInfo = rawMsg.getObject("osInfo", OsInfo.class);
				String agentId = rawMsg.getString("token");
				RenderAgent renderAgent = RenderAgent.dao.findById(agentId);
				if (renderAgent != null) {
					renderAgent.setCpuTotal(osInfo.getCpuTotal());
					renderAgent.setMemoryTotal(osInfo.getMemoryTotal());
					renderAgent.setMemoryFree(osInfo.getMemoryFree());
					renderAgent.setMemoryUsed(osInfo.getMemoryUsed());
					renderAgent.setDiskFree(osInfo.getDiskFree());
					renderAgent.setDiskUsable(osInfo.getDiskUsable());
					renderAgent.setDiskTotal(osInfo.getDiskTotal());
					renderAgent.setGpuTotal(osInfo.getGpuTotal());
					renderAgent.setCpuFreeUsageRate(osInfo.getCpuFreeUsageRate());
					renderAgent.setCpuSysUsageRate(osInfo.getCpuSysUsageRate());
					renderAgent.setCpuWaitingUsageRate(osInfo.getCpuWaitingUsageRate());
					renderAgent.setCpuUserUsageRate(osInfo.getCpuUserUsageRate());
					renderAgent.update();

					List<GpuInfo> gpuInfos = osInfo.getGpuInfos();
					for (GpuInfo gpuInfo : gpuInfos) {
						
						sb.append("'").append(gpuInfo.getBoardId()).append("',");
						
						RenderAgentGpu renderAgentGpu = RenderAgentGpu.dao.findByIds(gpuInfo.getBoardId(), agentId);
						if (renderAgentGpu == null) {
							renderAgentGpu = new RenderAgentGpu();
							renderAgentGpu.setId(gpuInfo.getBoardId());
							renderAgentGpu.setAgentId(agentId);
							renderAgentGpu.setName(gpuInfo.getProductName());
							renderAgentGpu.setMemoryTotal(gpuInfo.getMemoryTotal());
							renderAgentGpu.setMemoryUsed(gpuInfo.getMemoryUsed());
							renderAgentGpu.setMemoryFree(gpuInfo.getMemoryFree());
							renderAgentGpu.setIndex(Integer.parseInt(gpuInfo.getModuleId()));
							renderAgentGpu.setModifyTime(DateUtil.getLocalDateTime());
							renderAgentGpu.save();
						}else {
							renderAgentGpu.setName(gpuInfo.getProductName());
							renderAgentGpu.setIndex(Integer.parseInt(gpuInfo.getModuleId()));
							renderAgentGpu.setMemoryUsed(gpuInfo.getMemoryUsed());
							renderAgentGpu.setMemoryFree(gpuInfo.getMemoryFree());
							renderAgentGpu.setModifyTime(DateUtil.getLocalDateTime());
							renderAgentGpu.update();
						}
					}
					try {
						Db.delete("DELETE FROM "+RenderAgentGpu.TABLE_NAME+"  WHERE ID NOT IN ("+sb.substring(0,sb.length()-1).toString()+") AND AGENT_ID='"+agentId+"' ");
					}catch (Exception e) {
						e.printStackTrace();
					}
				}
			} 
			else if (StrKit.equals(type, "Stop_Order")) {
				//String agentToken=rawMsg.getString("token");
				String orderId=rawMsg.getString("orderId");
				RenderLog renderLog=RenderLog.dao.findById(orderId);
				if(renderLog!=null) {
					renderLog.setStatus("stop");
					renderLog.setMessage("停止渲染.");
					renderLog.setOfflineTime(new Date(System.currentTimeMillis()));
					renderLog.update();
				}

			}
			else {
				log.error("unsupported RenderNoder message type:%s", type);
			}
		}
	}

	/**
	 * 向客户端发送心跳
	 * 
	 * @param ctx
	 * @param evt
	 * @throws Exception
	 */
	@Override
	public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
		if (evt instanceof IdleStateEvent) {
			IdleStateEvent event = (IdleStateEvent) evt;
			if (event.state().equals(IdleState.READER_IDLE)) {
				sendPing(ctx, System.currentTimeMillis());
			} else if (event.state().equals(IdleState.WRITER_IDLE)) {
				sendPing(ctx, System.currentTimeMillis());
			} else if (event.state().equals(IdleState.ALL_IDLE)) {
				sendPing(ctx, System.currentTimeMillis());
			}
		}
	}

	public void send(ChannelHandlerContext ctx, String json) {
		if (ctx != null) {
			ctx.channel().writeAndFlush(new TextWebSocketFrame(json)).addListener(new ChannelFutureListener() {
				@Override
				public void operationComplete(ChannelFuture future) throws Exception {
					if (future.isSuccess()) {
						// log.info(String.format("-> Streamer:%s", json));
					} else {
						// log.info(String.format("-> Streamer isError :%s", json));
					}
				}
			});
		}
	}

	public void sendPing(ChannelHandlerContext ctx, long currentTimeMillis) {
		Map<String, Object> clientConfig = new HashMap<String, Object>();
		clientConfig.put("type", "pong");
		clientConfig.put("time", currentTimeMillis);
		send(ctx, JsonKit.toJson(clientConfig));
	}

	public void sendUnauthorized(ChannelHandlerContext ctx) {
		Map<String, Object> clientConfig = new HashMap<String, Object>();
		clientConfig.put("type", "unauthorized");
		clientConfig.put("message", "请先获取授权许可.");
		send(ctx, JsonKit.toJson(clientConfig));
	}

	public void sendNotAvailable(ChannelHandlerContext ctx) {
		Map<String, Object> clientConfig = new HashMap<String, Object>();
		clientConfig.put("type", "NotAvailable");
		clientConfig.put("message", "已禁用.");
		send(ctx, JsonKit.toJson(clientConfig));
	}

	public void sendConnected(String token,ChannelHandlerContext ctx) {
		Map<String, Object> clientConfig = new HashMap<String, Object>();
		clientConfig.put("type", "connected");
		clientConfig.put("message", "已接入.");
		send(ctx, JsonKit.toJson(clientConfig));
		
		RenderAgentSignalling renderAgentSignalling = RenderAgentSignalling.dao.findById(token);
		
		if(renderAgentSignalling!=null && IPUtil.isIp(renderAgentSignalling.getPlayerIp()) && IPUtil.isIp(renderAgentSignalling.getStreamerIp())) {
			Map<String, Object> signalling = new HashMap<String, Object>();
			signalling.put("type", "SynForSignalling");
			signalling.put("signalling", renderAgentSignalling);
			send(ctx, JsonKit.toJson(signalling));
		}
		

		
		RenderAgentRelay renderAgentRelay=RenderAgentRelay.dao.findById(token);
		if(renderAgentRelay!=null && IPUtil.isIp(renderAgentRelay.getIp())) {
			Map<String, Object> SFU = new HashMap<String, Object>();
			SFU.put("type", "SynForSFU");
			SFU.put("SFU", renderAgentRelay);
			send(ctx, JsonKit.toJson(SFU));

		}

		List<RenderAgentGpu> renderAgentGpus=RenderAgentGpu.dao.find("select a.id,a.index,a.limit from "+RenderAgentGpu.TABLE_NAME+" a where a.agent_id=?",token);
		List<Project> projects =Project.dao.find("select id,exe_relative_path from "+Project.TABLE_NAME+" where id in (select project_id from master_project_syn_status where agent_id=? )",token);
		Map<String, Object> Gpus = new HashMap<String, Object>();
		Gpus.put("type", "SynForGPU");
		Gpus.put("gpus", renderAgentGpus);
		Gpus.put("projects", projects);
		Gpus.put("agentId",token);
		send(ctx, JsonKit.toJson(Gpus));
	}
	

	public void onRenderDisconnected(ChannelHandlerContext ctx) {
		if (redisImpl.hasKey(ctx.channel().id().asLongText())) {
			String agentToken = redisImpl.get(ctx.channel().id().asLongText());
			if (!StrKit.isBlank(agentToken) && Cirrus.AgentManager.containsKey(agentToken)) {
				Cirrus.AgentManager.remove(agentToken);
				redisImpl.delete(agentToken);
				Set<String> Agent_GPU_Indexs = redisImpl.getByPrefix(agentToken);
				for (String GPUIndex : Agent_GPU_Indexs) {
					redisImpl.delete(redisImpl.get(GPUIndex));
				}
				redisImpl.delete(Agent_GPU_Indexs);
			}
			redisImpl.delete(ctx.channel().id().toString());

			Set<String> GraphicsAdaptersForFree = redisImpl.getByPrefix(PrefixUtil.GPU_Free);
			Set<String> GraphicsAdaptersForUsed = redisImpl.getByPrefix(PrefixUtil.GPU_Used);
			Set<String> GraphicsAdaptersForAgent = redisImpl.getByPrefix(agentToken);

			
			log.info(String.format(">>> GraphicsAdaptersForAgent{%s} is closed.", agentToken));
			log.info(String.format(">>> GraphicsAdaptersForFree:%s .", GraphicsAdaptersForFree.size()));
			log.info(String.format(">>> GraphicsAdaptersForUsed:%s .", GraphicsAdaptersForUsed.size()));
			log.info(String.format(">>> GraphicsAdaptersForAgent:{%s} - %s .", agentToken,GraphicsAdaptersForAgent.size()));

		}
	}

	public void onGPUInfo(ChannelHandlerContext ctx, JSONObject msg) {
		String agentToken = msg.getString("token");
		String oldKey = msg.getString("oldKey");
		String newKey = msg.getString("newKey");

		if (redisImpl.hasKey(oldKey)) {
			redisImpl.delete(String.format("%s_%s", agentToken, oldKey));
			redisImpl.delete(oldKey);
		}

		if(!redisImpl.hasKey(newKey)) {
			redisImpl.add(newKey, agentToken);
			redisImpl.add(String.format("%s_%s", agentToken, newKey), newKey);
		}

		Map<String, Object> clientConfig = new HashMap<String, Object>();
		clientConfig.put("type", "onGPUInfo");
		clientConfig.put("state", "success");
		clientConfig.put("message", "OnGPUInfoSuccess.");
		send(ctx, JsonKit.toJson(clientConfig));
		
		Set<String> GraphicsAdaptersForFree = redisImpl.getByPrefix(PrefixUtil.GPU_Free);
		Set<String> GraphicsAdaptersForUsed = redisImpl.getByPrefix(PrefixUtil.GPU_Used);
		Set<String> GraphicsAdaptersForAgent = redisImpl.getByPrefix(agentToken);

		
		log.info(String.format(">>> GraphicsAdaptersForFree:%s.", GraphicsAdaptersForFree.size()));
		log.info(String.format(">>> GraphicsAdaptersForUsed:%s.", GraphicsAdaptersForUsed.size()));
		log.info(String.format(">>> GraphicsAdaptersForAgent:{%s} - %s.", agentToken, GraphicsAdaptersForAgent.size()));

	}

	public void onGPUInit(ChannelHandlerContext ctx, JSONObject msg) {

		String agentToken = msg.getString("token");
		String index = msg.getString("index");
		String uuid = msg.getString("uuid");
		int sn = msg.getInteger("sn");

		String rFreeKey = KeyUtil.getFreeKey(index, uuid, sn);
		redisImpl.add(rFreeKey, agentToken);
		redisImpl.add(String.format("%s_%s", agentToken, rFreeKey), rFreeKey);

		Map<String, Object> clientConfig = new HashMap<String, Object>();
		clientConfig.put("type", "onGPUInit");
		clientConfig.put("state", "success");
		clientConfig.put("message", "OnGPUInitSuccess.");
		send(ctx, JsonKit.toJson(clientConfig));
	}

}
